#include "../test_helpers.h"
#include "MessageProducer.h"
#include <spdlog/spdlog.h>
#include "MessageConsumer.h"
#include <gtest/gtest.h>
#include <thread>
#include <atomic>

using namespace MessageFramework;
using namespace MessageFramework::Test;

class FaultToleranceTest : public RedisTestBase {
protected:
    std::string streamName;
    std::string consumerGroup;

    void SetUp() override {
        RedisTestBase::SetUp();
        streamName = testStreamPrefix + "_fault";
        consumerGroup = testStreamPrefix + "_group";
        spdlog::set_level(spdlog::level::warn);  // 减少日志噪音
    }
};

// ==================== Consumer崩溃恢复测试 ====================

TEST_F(FaultToleranceTest, ConsumerCrashRecovery) {
    // 发送5条消息
    MessageProducer::Config producerConfig;
    producerConfig.publisherId = "TestProducer";
    MessageProducer producer(redisPool, producerConfig);

    for (int i = 0; i < 5; ++i) {
        std::string msgId = producer.send(streamName, "TestMessage", "{\"index\":" + std::to_string(i) + "}");
        std::cout << "发送消息 " << i << ": " << (msgId.empty() ? "失败" : "成功 ID=" + msgId) << std::endl;
    }

    std::cout << "所有消息发送完成，streamName=" << streamName << std::endl;

    // 启动第一个消费者，通过callback限制只处理2条
    std::atomic<int> processedCount{0};
    std::shared_ptr<MessageConsumer> consumer1;

    {
        MessageConsumer::Config consumerConfig;
        consumerConfig.streamName = streamName;
        consumerConfig.consumerGroup = consumerGroup;
        consumerConfig.consumerId = "Consumer1";
        consumerConfig.autoCreateGroup = true;
        consumerConfig.groupStartId = "0";  // 从头开始读取所有消息
        consumerConfig.batchSize = 10;  // 一次读取多条
        consumerConfig.filterByConsumerId = false;  // 不过滤consumerId，处理所有消息

        consumer1 = std::make_shared<MessageConsumer>(redisPool, consumerConfig);

        // 使用callback模式，处理2条后立即停止Consumer
        consumer1->setMessageCallback([&](const Message& msg) -> int {
            if (processedCount >= 2) {
                // 已经处理够了，停止消费者（在另一个线程中停止以避免死锁）
                std::thread([consumer1]() {
                    std::this_thread::sleep_for(std::chrono::milliseconds(10));
                    consumer1->stop();
                }).detach();
                return 0;  // 仍然返回成功
            }
            processedCount++;
            std::cout << "Consumer1处理消息 " << processedCount.load() << std::endl;
            return 0;
        });

        consumer1->start();
        std::cout << "Consumer1已启动，等待处理消息..." << std::endl;

        // 等待处理2条消息并自动停止
        TestHelpers::waitFor([&]() {
            return processedCount >= 2;
        }, 5000);

        // 给Consumer一些时间停止
        std::this_thread::sleep_for(std::chrono::milliseconds(500));

        auto stats1 = consumer1->getStats();
        std::cout << "Consumer1 stats: received=" << stats1.messagesReceived
                  << " processed=" << stats1.messagesProcessed << std::endl;

        // 模拟崩溃（不正常停止，不调用stop()）
        // Consumer应该已经自动停止，剩余的3条消息未被处理
    }

    std::this_thread::sleep_for(std::chrono::seconds(1));

    // 启动第二个消费者接管
    MessageConsumer::Config consumerConfig2;
    consumerConfig2.streamName = streamName;
    consumerConfig2.consumerGroup = consumerGroup;
    consumerConfig2.consumerId = "Consumer2";
    consumerConfig2.autoCreateGroup = false;  // Group已存在
    consumerConfig2.filterByConsumerId = false;  // 不过滤consumerId，处理所有消息
    consumerConfig2.pendingCheckInterval = 1;  // 1秒检查一次pending
    consumerConfig2.pendingTimeout = 1;  // 1秒后认为消息超时

    auto dispatcher2 = std::make_shared<MessageDispatcher>();
    auto handler2 = std::make_shared<MockHandler>("TestMessage");
    dispatcher2->registerHandler("TestMessage", handler2);

    MessageConsumer consumer2(redisPool, consumerConfig2, dispatcher2);
    consumer2.start();

    // 等待pending消息被reclaim和处理（需要等待pendingCheckInterval触发）
    TestHelpers::waitFor([&]() {
        return handler2->getCallCount() >= 3;  // 至少处理剩余的3条
    }, 15000);  // 增加到15秒，给pending check足够时间

    // 验证总消息数
    auto stats = consumer2.getStats();
    EXPECT_GE(stats.messagesReceived, 3);

    consumer2.stop();
}

// ==================== 多消费者负载均衡测试 ====================

TEST_F(FaultToleranceTest, MultiConsumerLoadBalancing) {
    // 发送20条消息
    MessageProducer::Config producerConfig;
    producerConfig.publisherId = "TestProducer";
    MessageProducer producer(redisPool, producerConfig);

    for (int i = 0; i < 20; ++i) {
        producer.send(streamName, "TestMessage", "{\"index\":" + std::to_string(i) + "}");
    }

    // 创建2个消费者
    MessageConsumer::Config config1;
    config1.streamName = streamName;
    config1.consumerGroup = consumerGroup;
    config1.consumerId = "Consumer1";
    config1.batchSize = 5;
    config1.autoCreateGroup = true;

    auto dispatcher1 = std::make_shared<MessageDispatcher>();
    auto handler1 = std::make_shared<MockHandler>("TestMessage");
    dispatcher1->registerHandler("TestMessage", handler1);

    MessageConsumer consumer1(redisPool, config1, dispatcher1);

    MessageConsumer::Config config2;
    config2.streamName = streamName;
    config2.consumerGroup = consumerGroup;
    config2.consumerId = "Consumer2";
    config2.batchSize = 5;
    config2.autoCreateGroup = false;

    auto dispatcher2 = std::make_shared<MessageDispatcher>();
    auto handler2 = std::make_shared<MockHandler>("TestMessage");
    dispatcher2->registerHandler("TestMessage", handler2);

    MessageConsumer consumer2(redisPool, config2, dispatcher2);

    // 同时启动
    consumer1.start();
    consumer2.start();

    // 等待所有消息被处理
    TestHelpers::waitFor([&]() {
        return (handler1->getCallCount() + handler2->getCallCount()) >= 20;
    }, 15000);

    int total = handler1->getCallCount() + handler2->getCallCount();
    EXPECT_EQ(total, 20);

    // 验证负载均衡（每个消费者都应该处理一些消息）
    EXPECT_GT(handler1->getCallCount(), 0);
    EXPECT_GT(handler2->getCallCount(), 0);

    std::cout << "\n负载均衡测试结果：" << std::endl;
    std::cout << "  Consumer1处理: " << handler1->getCallCount() << " 条" << std::endl;
    std::cout << "  Consumer2处理: " << handler2->getCallCount() << " 条" << std::endl;
    std::cout << "  总计: " << total << " 条" << std::endl;

    consumer1.stop();
    consumer2.stop();
}

// ==================== 签名验证失败处理测试 ====================

TEST_F(FaultToleranceTest, SignatureVerificationFailure) {
    std::string secretKey = "test-secret-key-32bytes-long!!!";

    // 创建消费者（启用签名验证）
    MessageConsumer::Config consumerConfig;
    consumerConfig.streamName = streamName;
    consumerConfig.consumerGroup = consumerGroup;
    consumerConfig.consumerId = "TestConsumer";

    auto dispatcher = std::make_shared<MessageDispatcher>();
    MessageDispatcher::Config dispatcherConfig;
    dispatcherConfig.secretKey = secretKey;
    dispatcherConfig.verifySignature = true;
    dispatcher->setConfig(dispatcherConfig);

    auto handler = std::make_shared<MockHandler>("TestMessage");
    dispatcher->registerHandler("TestMessage", handler);

    MessageConsumer consumer(redisPool, consumerConfig, dispatcher);
    consumer.start();

    // 发送有效签名的消息
    MessageProducer::Config validProducerConfig;
    validProducerConfig.publisherId = "ValidProducer";
    validProducerConfig.secretKey = secretKey;
    validProducerConfig.autoSign = true;

    MessageProducer validProducer(redisPool, validProducerConfig);
    validProducer.send(streamName, "TestMessage", "{\"valid\":true}");

    // 发送无效签名的消息（使用错误密钥）
    MessageProducer::Config invalidProducerConfig;
    invalidProducerConfig.publisherId = "InvalidProducer";
    invalidProducerConfig.secretKey = "wrong-key-32bytes-long!!!!!!!";
    invalidProducerConfig.autoSign = true;

    MessageProducer invalidProducer(redisPool, invalidProducerConfig);
    invalidProducer.send(streamName, "TestMessage", "{\"valid\":false}");

    // 发送另一条有效消息
    validProducer.send(streamName, "TestMessage", "{\"valid\":true}");

    // 等待消息被处理
    std::this_thread::sleep_for(std::chrono::seconds(3));

    // 应该只处理2条有效消息
    EXPECT_EQ(handler->getCallCount(), 2);

    auto stats = consumer.getStats();
    EXPECT_EQ(stats.messagesFailed, 1);  // 1条签名验证失败

    consumer.stop();
}

// ==================== Handler异常处理测试 ====================

TEST_F(FaultToleranceTest, HandlerException) {
    // 创建会抛出异常的Handler
    class ExceptionHandler : public IMessageHandler {
    public:
        std::atomic<int> callCount{0};

        int handleMessage(const Message& message) override {
            callCount++;
            if (callCount == 2) {
                throw std::runtime_error("Handler exception");
            }
            return 0;
        }

        std::string getMessageType() const override { return "TestMessage"; }
        std::string getName() const override { return "ExceptionHandler"; }
    };

    MessageConsumer::Config consumerConfig;
    consumerConfig.streamName = streamName;
    consumerConfig.consumerGroup = consumerGroup;
    consumerConfig.consumerId = "TestConsumer";

    auto dispatcher = std::make_shared<MessageDispatcher>();
    auto handler = std::make_shared<ExceptionHandler>();
    dispatcher->registerHandler("TestMessage", handler);

    MessageConsumer consumer(redisPool, consumerConfig, dispatcher);
    consumer.start();

    // 发送3条消息
    MessageProducer::Config producerConfig;
    producerConfig.publisherId = "TestProducer";
    MessageProducer producer(redisPool, producerConfig);

    for (int i = 0; i < 3; ++i) {
        producer.send(streamName, "TestMessage", "{\"index\":" + std::to_string(i) + "}");
    }

    // 等待所有消息被处理
    TestHelpers::waitFor([&]() {
        return handler->callCount >= 3;
    }, 5000);

    // 验证所有消息都被调用（即使第2条抛出异常）
    EXPECT_EQ(handler->callCount, 3);

    auto stats = consumer.getStats();
    EXPECT_EQ(stats.messagesFailed, 1);  // 第2条消息处理失败

    consumer.stop();
}

int main(int argc, char** argv) {
    ::testing::InitGoogleTest(&argc, argv);
    return RUN_ALL_TESTS();
}
